Purpose

Use Google Cloud Storage (GCS) as warehouse / object storage for pypaimon (without java dependencies). This PR adds PyArrowFileIO handling for GCS.

Changes

1. Scheme dispatch (pyarrow_file_io.py)

Added a gs branch in __init__ alongside the existing s3 and hdfs branches:

elif scheme == "gs":
    self.filesystem = self._initialize_gcs_fs()

2. _initialize_gcs_fs() method (pyarrow_file_io.py)

Uses pyarrow.fs.GcsFileSystem, which is already available via pyarrow[gcs].

With no arguments, GcsFileSystem picks up credentials automatically via Application Default Credentials (ADC): GOOGLE_APPLICATION_CREDENTIALS, the GCP metadata server, or Workload Identity on GKE/GCE. Three optional properties allow explicit credential passing:

def _initialize_gcs_fs(self) -> FileSystem:
    access_token = self._get_property("gcs.access-token")
    token_expiry = self._get_property("gcs.access-token.expiration")
    project_id = self._get_property("gcs.project-id")

    kwargs = {}
    if access_token:
        from datetime import datetime
        kwargs["access_token"] = access_token
        kwargs["credential_token_expiration"] = (
            datetime.fromisoformat(token_expiry) if token_expiry
            else datetime(9999, 12, 31)
        )
    if project_id:
        kwargs["project_id"] = project_id

    # With no kwargs, GcsFileSystem uses ADC automatically
    return pafs.GcsFileSystem(**kwargs)

3. Path fix in to_filesystem_path() (pyarrow_file_io.py)

to_filesystem_path() had a catch-all for non-S3/non-HDFS schemes that returned only uri.path — e.g. for gs://my-bucket/data/table it returned /data/table, stripping the bucket name. GcsFileSystem (like S3FileSystem) expects paths in bucket/key form without a leading slash.

Added a GcsFileSystem branch mirroring the existing S3 logic:

from pyarrow.fs import GcsFileSystem
if isinstance(self.filesystem, GcsFileSystem):
    if parsed.scheme and parsed.netloc:
        path_part = normalized_path.lstrip('/')
        return f"{parsed.netloc}/{path_part}" if path_part else parsed.netloc
    return str(path)

4. GcsOptions class (config.py)

Added a GcsOptions class alongside the existing S3Options documenting the three new properties.

class GcsOptions:
    GCS_ACCESS_TOKEN = ConfigOptions.key("gcs.access-token").string_type().no_default_value().with_description(
        "GCS access token. If not set, ADC (Application Default Credentials) is used automatically.")
    ...

Linked issue

#7768

Tests

Unit tests — pypaimon/tests/file_io_test.py

A new GCSFileIOPathTest class is added to the existing file_io_test.py, following the same pattern as the existing S3/OSS path conversion tests. These tests require no GCS credentials.

python -m pytest pypaimon/tests/file_io_test.py::GCSFileIOPathTest -v

Integration tests — pypaimon/tests/gcs_file_io_test.py

A new GCSFileIOTest class is added following the pattern of oss_file_io_test.py. All tests are skipped automatically when GCS_TEST_BUCKET is not set. Credentials are picked up via ADC — no explicit credential properties are required.

# Skips all tests (no bucket configured)
python -m pytest pypaimon/tests/gcs_file_io_test.py -v

# Runs all tests against a real GCS bucket using ADC
GCS_TEST_BUCKET=my-bucket python -m pytest pypaimon/tests/gcs_file_io_test.py -v